package com.max.apptpl.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;


public class JWordcountBase {
    // MR 实现wordcount
    public static void main(String[] args) throws Exception {
        /**
         * 1 获取配置信息以及封装任务
         * 2 设置 jar 加载路径
         * 3 设置 map 和 reduce 类
         * 4 设置 map 输出
         * 5 设置最终输出 kv 类型
         * 6 设置输入和输出路径
         * 7 提交
         */
        Configuration config = new Configuration();
        Job instance = Job.getInstance(config, "word count");
        instance.setJarByClass(JWordcountBase.class);   // 可以改写成获取本文件类名

        instance.setMapperClass(TokenizerMapper.class);
        instance.setReducerClass(IntSumReducer.class);

        instance.setMapOutputKeyClass(Text.class);
        instance.setMapOutputValueClass(IntWritable.class);
        instance.setOutputKeyClass(Text.class);
        instance.setOutputValueClass(IntWritable.class);

//        FileInputFormat.setInputPaths(instance, new Path("log/wordcount/in/words.txt"));
        FileInputFormat.setInputPaths(instance, new Path(args[0]));
//        FileOutputFormat.setOutputPath(instance, new Path("log/wordcount/out"));
        FileOutputFormat.setOutputPath(instance, new Path(args[1]));

        boolean res = instance.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }

    // mapper
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

    // reducer
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values, Context context )
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

}


